﻿#include "RpcConnection.h"
#include "XyNet/Common/Logger.h"
#include "XyNet/Channel/RpcSocketChannel.h"
#include "XyNet/Endpoint/RpcEndpoint.h"
#include "XyNet/Protocol/cpp/Rpc.service.h"
#include "XyNet/Service/RpcController.h"

namespace XyNet
{
	RpcConnection::RpcConnection(RpcSocketChannel* socketChannel)
		: _weakSocketChannel(socketChannel->shared_from_this())
		, _strand(socketChannel->getSocket()->get_io_service())
		, _readTimer(make_unique<boost::asio::deadline_timer>(_strand.get_io_service()))
		, _writeTimer(make_unique<boost::asio::deadline_timer>(_strand.get_io_service()))
		, _readMessageCountPerSecond(0), _writeMessageCountPerSecond(0)
		, _maxReadMessageCountPerSecond(0), _maxWriteMessageCountPerSecond(0)
		, _readBytesPerSecond(0), _writeBytesPerSecond(0)
		, _maxReadBytesPerSecond(0), _maxWriteBytesPerSecond(0)
		, _connectComplete(false) {}

	RpcConnection::~RpcConnection() {}

	void RpcConnection::start()
	{
		asyncWrite("X");	// 随意写个字符告诉对方连接完成
		asyncRead(bind(&RpcConnection::handleReadKey, shared_from_this(), std::placeholders::_1));

		// 开始秒计时。简化处理，避免多次重连后计时器的事件叠加
		if (!_secondTimer)
		{
			_secondTimer = make_unique<boost::asio::deadline_timer>(_strand.get_io_service());
			startSecondTimer();
		}
	}

	void RpcConnection::write(const Bytes& data)
	{
		assert(!data.empty());
		if (isConnectComplete())
		{
			_strand.get_io_service().post(
				_strand.wrap(bind(&RpcConnection::asyncWrite, shared_from_this(), data)));
		}
		else
			pushCacheWriteData(data);
	}

	void RpcConnection::read()
	{
		asyncRead(bind(&RpcConnection::socketChannelHandleConnectionData,
			_weakSocketChannel, std::placeholders::_1));
	}

	shared_ptr<RpcSocketChannel> RpcConnection::getSocketChannel()
	{
		//return _endpoint->findSocketChannel(_socketChannelId);
		return _weakSocketChannel.lock();
	}

	void RpcConnection::pushCacheWriteData(const Bytes& data)
	{
		_cachedWriteData.emplace_back(data);
	}

	void RpcConnection::setConnectComplete()
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		_connectComplete = true;

		auto remoteEndpoint = socketChannel->getSocket()->remote_endpoint();
		boost::format format("%s:%s Connect complete.");
		format % remoteEndpoint.address() % remoteEndpoint.port();
		Logger::getDefault()->log(format.str());

		// 写入缓存数据
		for (const Bytes& data : _cachedWriteData)
			write(data);
		_cachedWriteData.clear();
	}

	void RpcConnection::asyncWrite(const Bytes& data)
	{
		bool writeInProgress = !_writeStreamDeque.empty();
		_writeStreamDeque.emplace_back(data);
		if (!writeInProgress)	// 如果没有在写的消息，直接开始写处理，一次只能一个writeNext
			writeNext();
	}

	void RpcConnection::asyncRead(const DataHandler& handler)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		boost::asio::async_read(*socketChannel->getSocket(),
			boost::asio::buffer(_readStream.getHeadPtr(), _readStream.getHeadSize()),
			_strand.wrap(bind(&RpcConnection::handleReadSize, shared_from_this(),
				boost::asio::placeholders::error, handler)));
	}

	void RpcConnection::writeNext()
	{
		if (!_writeStreamDeque.empty())
		{
			// 检测发送消息数量
			auto socketChannel = getSocketChannel();
			if (socketChannel)
			{
				auto config = socketChannel->getEndpoint()->getConfig();
				if (_writeMessageCountPerSecond >= config->getMaxMessageCountPerSecond())
				{
					_writeTimer->expires_from_now(boost::posix_time::seconds(1));
					_writeTimer->async_wait(_strand.wrap(bind(&RpcConnection::handleWriteTimer,
						shared_from_this(), boost::asio::placeholders::error)));
				}
				else
				{
					++_writeMessageCountPerSecond;

					_writeStream.swap(_writeStreamDeque.front());
					if (_writeStream.getDataSize() > config->getMaxMessagePacketSize())
					{
						socketChannel->handleSocketChannelException(
							RpcErrors::ExceedMaxMessagePacketSize);
					}
					else
					{
						writeSegment(0);
					}
				}
			}
		}
	}

	void RpcConnection::readNext()
	{
		// 检测接收消息数量
		auto socketChannel = getSocketChannel();
		if (socketChannel)
		{
			auto config = socketChannel->getEndpoint()->getConfig();
			if (_readMessageCountPerSecond >= config->getMaxMessageCountPerSecond())
			{
				_readTimer->expires_from_now(boost::posix_time::seconds(1));
				_readTimer->async_wait(_strand.wrap(bind(&RpcConnection::handleReadTimer,
					shared_from_this(), boost::asio::placeholders::error)));
			}
			else
			{
				++_readMessageCountPerSecond;
				read();
			}
		}
	}

	void RpcConnection::handleWrite(const boost::system::error_code& error)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		if (error)
			socketChannel->handleSocketException(error);
		else
		{
			_writeStreamDeque.pop_front();	// 弹出写完后的流，好开始下一步
			// 只有连接准备完毕（如交换好加密密钥时），才能自动开始下一次写入
			if (_connectComplete)
				writeNext();
		}
	}

	void RpcConnection::writeSegment(int offset)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		auto maxUploadSpeed = socketChannel->getEndpoint()->getConfig()->getMaxUploadSpeed();
		auto canWriteSize = maxUploadSpeed - _writeBytesPerSecond;
		auto needWriteSize = _writeStream.getDataSize() - offset;
		if (canWriteSize < needWriteSize)
		{
			_writeBytesPerSecond += canWriteSize;
			boost::asio::async_write(*socketChannel->getSocket(),
				boost::asio::buffer(_writeStream.getData() + offset, canWriteSize),
				_strand.wrap(bind(&RpcConnection::handleWriteSegment, shared_from_this(),
					boost::asio::placeholders::error, offset + canWriteSize)));
		}
		else
		{
			_writeBytesPerSecond += needWriteSize;
			boost::asio::async_write(*socketChannel->getSocket(),
				boost::asio::buffer(_writeStream.getData() + offset, needWriteSize),
				_strand.wrap(bind(&RpcConnection::handleWrite, shared_from_this(),
					boost::asio::placeholders::error)));
		}
	}

	void RpcConnection::handleWriteSegment(const boost::system::error_code& error, int offset)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		if (error)
			socketChannel->handleSocketException(error);
		else
		{
			_writeTimer->expires_from_now(boost::posix_time::seconds(1));
			_writeTimer->async_wait(_strand.wrap(bind(&RpcConnection::handleWriteTimer,
				shared_from_this(), boost::asio::placeholders::error, offset)));
		}
	}

	void RpcConnection::handleWriteTimer(weak_ptr<RpcConnection> weakConnection,
		const boost::system::error_code& error)
	{
		if (!error)
		{
			auto connection = weakConnection.lock();
			if (connection)
				connection->writeNext();
		}
	}

	void RpcConnection::handleWriteTimer(weak_ptr<RpcConnection> weakConnection,
		const boost::system::error_code& error, int offset)
	{
		if (!error)
		{
			auto connection = weakConnection.lock();
			if (connection)
				connection->writeSegment(offset);
		}
	}

	void RpcConnection::handleReadSize(
		const boost::system::error_code& error, const DataHandler& handler)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		if (error)
			socketChannel->handleSocketException(error);
		else
		{
			auto config = socketChannel->getEndpoint()->getConfig();
			if (!_readStream.allocateBodyByHead())	// 无效尺寸
				socketChannel->handleSocketChannelException(RpcErrors::InvalidMessage);
			else if (_readStream.getDataSize() > config->getMaxMessagePacketSize())
			{
				socketChannel->handleSocketChannelException(
					RpcErrors::ExceedMaxMessagePacketSize);
			}
			else
			{
				readSegment(0, handler);
			}
		}
	}

	void RpcConnection::handleReadData(
		const boost::system::error_code& error, const DataHandler& handler)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		if (error)
			socketChannel->handleSocketException(error);
		else
		{
			if (handler)
			{
				handler(Bytes(
					(const char*)_readStream.getBodyPtr(), _readStream.getBodySize()));
			}

			// 只有连接准备完毕（如交换好加密密钥时），才能自动开始下一次读取
			if (_connectComplete)
				readNext();
		}
	}

	void RpcConnection::readSegment(int offset, const DataHandler& handler)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		auto maxDownloadSpeed = socketChannel->getEndpoint()->getConfig()->getMaxDownloadSpeed();
		auto canReadSize = maxDownloadSpeed - _readBytesPerSecond;
		auto needReadSize = _readStream.getBodySize() - offset;
		if (canReadSize < needReadSize)
		{
			_readBytesPerSecond += canReadSize;
			boost::asio::async_read(*socketChannel->getSocket(),
				boost::asio::buffer(_readStream.getBodyPtr() + offset, canReadSize),
				_strand.wrap(bind(&RpcConnection::handleReadSegment, shared_from_this(),
					boost::asio::placeholders::error, offset + canReadSize, handler)));
		}
		else
		{
			_readBytesPerSecond += needReadSize;
			boost::asio::async_read(*socketChannel->getSocket(),
				boost::asio::buffer(_readStream.getBodyPtr() + offset, needReadSize),
				_strand.wrap(bind(&RpcConnection::handleReadData, shared_from_this(),
					boost::asio::placeholders::error, handler)));
		}
	}

	void RpcConnection::handleReadSegment(
		const boost::system::error_code& error, int offset, const DataHandler& handler)
	{
		auto socketChannel = getSocketChannel();
		if (!socketChannel)
			return;

		if (error)
			socketChannel->handleSocketException(error);
		else
		{
			_readTimer->expires_from_now(boost::posix_time::seconds(1));
			_readTimer->async_wait(_strand.wrap(bind(&RpcConnection::handleReadTimer,
				shared_from_this(), boost::asio::placeholders::error, offset, handler)));
		}
	}

	void RpcConnection::handleReadTimer(weak_ptr<RpcConnection> weakConnection,
		const boost::system::error_code& error)
	{
		if (!error)
		{
			auto connection = weakConnection.lock();
			if (connection)
				connection->readNext();
		}
	}

	void RpcConnection::handleReadTimer(weak_ptr<RpcConnection> weakConnection,
		const boost::system::error_code& error, int offset, const DataHandler& handler)
	{
		if (!error)
		{
			auto connection = weakConnection.lock();
			if (connection)
				connection->readSegment(offset, handler);
		}
	}

	void RpcConnection::handleReadKey(const Bytes& data)
	{
		setConnectComplete();	// 设置准备完成后自动开始读循环
	}

	void RpcConnection::startSecondTimer()
	{
		_secondTimer->expires_from_now(boost::posix_time::seconds(1));
		_secondTimer->async_wait(_strand.wrap(bind(&RpcConnection::handleSecondTimer,
			shared_from_this(), boost::asio::placeholders::error)));
	}

	void RpcConnection::handleSecondTimer(weak_ptr<RpcConnection> weakConnection, const boost::system::error_code& error)
	{
		if (!error)
		{
			auto connection = weakConnection.lock();
			if (connection)
			{
				if (connection->_readMessageCountPerSecond >
					connection->_maxReadMessageCountPerSecond)
				{
					connection->_maxReadMessageCountPerSecond =
						connection->_readMessageCountPerSecond;
				}
				connection->_readMessageCountPerSecond = 0;
				if (connection->_writeMessageCountPerSecond >
					connection->_maxWriteMessageCountPerSecond)
				{
					connection->_maxWriteMessageCountPerSecond =
						connection->_writeMessageCountPerSecond;
				}
				connection->_writeMessageCountPerSecond = 0;

				if (connection->_readBytesPerSecond > connection->_maxReadBytesPerSecond)
					connection->_maxReadBytesPerSecond = connection->_readBytesPerSecond;
				connection->_readBytesPerSecond = 0;
				if (connection->_writeBytesPerSecond > connection->_maxWriteBytesPerSecond)
					connection->_maxWriteBytesPerSecond = connection->_writeBytesPerSecond;
				connection->_writeBytesPerSecond = 0;

				connection->startSecondTimer();
			}
		}
	}

	void RpcConnection::socketChannelHandleConnectionData(weak_ptr<RpcSocketChannel> weakSocketChannel, const Bytes& data)
	{
		auto socketChannel = weakSocketChannel.lock();
		if (socketChannel)
			socketChannel->handleConnectionData(data);
	}
}